package com.zm.xiaohashu.count.biz.consumer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.github.phantomthief.collection.BufferTrigger;
import com.zm.xiaohashu.count.biz.constant.MQConstants;
import com.zm.xiaohashu.count.biz.constant.RedisKeyConstants;
import com.zm.xiaohashu.count.biz.enums.FollowUnfollowTypeEnum;
import com.zm.xiaohashu.count.biz.model.dto.CountFollowUnfollowMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_COUNT_FANS, // Group 组
        topic = MQConstants.TOPIC_COUNT_FANS // 主题 Topic
)
@Slf4j
public class CountFansConsumer implements RocketMQListener<String> { // TODO: Message

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000) // 缓存队列的最大容量
            .batchSize(1000)   // 一批次最多聚合 1000 条
            .linger(Duration.ofSeconds(1)) // 多久聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;


    @Override
    public void onMessage(String body) {
        log.info("## 消费到了 MQ 【计数: 粉丝数】, {}...", body);
        if (StrUtil.isBlank(body)) return;
        bufferTrigger.enqueue(body);
    }

    private void consumeMessage(List<String> body) {
        //userId---List<>
        Map<Long, List<CountFollowUnfollowMqDTO>> map = body.stream().map(v -> JSONUtil.toBean(v, CountFollowUnfollowMqDTO.class)).collect(Collectors.groupingBy(CountFollowUnfollowMqDTO::getTargetUserId));
        HashMap<Long, Integer> countMap = new HashMap<>();//落库用
        for (Map.Entry<Long, List<CountFollowUnfollowMqDTO>> entry : map.entrySet()) {
            Integer totalCount = 0;
            Long userId = entry.getKey();
            List<CountFollowUnfollowMqDTO> list = entry.getValue();
            for (CountFollowUnfollowMqDTO countFollowUnfollowMqDTO : list) {
                Integer type = countFollowUnfollowMqDTO.getType();
                totalCount += Objects.equals(type, FollowUnfollowTypeEnum.FOLLOW.getCode()) ? 1 : -1;
            }
            countMap.put(userId, totalCount);
        }
        countMap.forEach((userId, totalCount) -> {
            String userCountKey = RedisKeyConstants.buildCountUserKey(userId);
            boolean isExisted = redisTemplate.hasKey(userCountKey);
            if (isExisted) {
                redisTemplate.opsForHash().increment(userCountKey, RedisKeyConstants.FIELD_FANS_TOTAL, totalCount);
            }
        });
        //mq落库
        Message<HashMap<Long, Integer>> message = MessageBuilder.withPayload(countMap).build();
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS_2_DB, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：粉丝数入库】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：粉丝数入库】MQ 发送异常: ", throwable);
            }
        });
    }


}